//package com.hainiu.cat.web.kafka.config;
//
//import org.apache.kafka.clients.admin.NewTopic;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
//import org.springframework.kafka.core.ConsumerFactory;
//import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
//
//import javax.annotation.Resource;
//
///**
// * create by biji.zhao on 2021/6/29
// *
// * 执行代码kafkaTemplate.send("topic1", normalMessage)发送消息时，
// * kafka会帮我们自动完成topic的创建工作，
// * 但这种情况下创建的topic默认只有一个分区，分区也没有副本
// */
//@Configuration
//public class KafkaInitialConfiguration<K, V> {
//
//    @Resource
//    private ConsumerFactory<K, V> consumerFactory;
//
//    // 创建一个名为testtopic的Topic并设置分区数为8，分区副本数为2
//    @Bean
//    public NewTopic initialTopic() {
//        return new NewTopic("testtopic",8, (short) 2 );
//    }
//
//    // 如果要修改分区数，只需修改配置值重启项目即可
//    // 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
//    @Bean
//    public NewTopic updateTopic() {
//        return new NewTopic("testtopic",10, (short) 2 );
//    }
//
//    // 新建一个异常处理器，用@Bean注入
//    @Bean
//    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
//        return (message, exception, consumer) -> {
//            System.out.println("消费异常："+message.getPayload());
//            return null;
//        };
//    }
//
//    // 消息过滤器
//    @Bean
//    public ConcurrentKafkaListenerContainerFactory<K, V> filterContainerFactory() {
//        ConcurrentKafkaListenerContainerFactory<K, V> factory = new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(consumerFactory);
//        // 被过滤的消息将被丢弃
//        factory.setAckDiscarded(true);
//        // 消息过滤策略   返回true消息则被过滤
//        factory.setRecordFilterStrategy(consumerRecord -> !consumerRecord.value().toString().equals("海牛根"));
//        return factory;
//    }
//
//    // 监听器容器工厂(设置禁止KafkaListener自启动)
//    @Bean
//    public ConcurrentKafkaListenerContainerFactory<K, V> delayContainerFactory() {
//        ConcurrentKafkaListenerContainerFactory<K, V> container = new ConcurrentKafkaListenerContainerFactory<>();
//        container.setConsumerFactory(consumerFactory);
//        // 禁止KafkaListener自启动
//        container.setAutoStartup(false);
//        return container;
//    }
//}
